package org.example.sensor.service;

import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.example.sensor.dox.SensorData;
import org.example.sensor.repository.SensorDataRepository;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;

import java.time.LocalDateTime;

@Service
@RequiredArgsConstructor
@Slf4j
public class SensorDataHandler {

    private final SensorDataRepository repository;
    private final Sinks.Many<SensorData> dataSink = Sinks.many().multicast().onBackpressureBuffer();

    @ServiceActivator(inputChannel = "mqttInputChannel")
    public void handleMessage(Message<?> message) {
        try {
            String payload = message.getPayload().toString();
            SensorData data = parsePayload(payload);
            repository.save(data);

            // **🔍 先检查数据是否正确**
            SensorData sensorData = repository.findById(data.getId()).orElse(null);
            log.info("📦 数据库查询结果: {}", sensorData);

            // **✅ 发送到 SSE**
            if (sensorData != null) {
                log.info("🚀 推送 SSE 数据: {}", sensorData);
                Sinks.EmitResult result = dataSink.tryEmitNext(sensorData);
                log.info("📡 SSE 推送结果: {}", result);
            } else {
                log.warn("⚠️ 数据未找到，无法推送 SSE");
            }

        } catch (Exception e) {
            System.err.println("数据处理失败: " + e.getMessage());
        }
    }

    private SensorData parsePayload(String payload) {
        JsonObject json = JsonParser.parseString(payload).getAsJsonObject();
        return SensorData.builder()
                .temp(json.get("Temp").getAsDouble())
                .hum(json.get("Hum").getAsDouble())
                .light(json.get("Light").getAsInt())
                .smoke(json.get("Smoke").getAsInt())
                .build();
    }
    // 获取数据流（被 Controller 调用）
    public Flux<SensorData> getDataStream() {
        return dataSink.asFlux()
                .doOnSubscribe(subscription -> log.info("⚡ SSE 连接建立"))
                .doOnNext(data -> log.info("📡 发送 SSE 数据: {}", data))
                .doOnError(error -> log.error("🔥 SSE 发生错误: {}", error.getMessage()))
                .onErrorResume(error -> {
                    // 创建一个包含错误信息的 SensorData 对象
                    SensorData errorData = SensorData.builder()
                            .id("error-id")
                            .temp(-1)
                            .hum(-1)
                            .light(-1)
                            .smoke(-1)
                            .timestamp(LocalDateTime.now())
                            .build();
                    return Flux.just(errorData);
                });
    }
}